package com.yjh;

import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

//这个类是将mysql的数据转移到别一个mysql中
//mysql------>flinkCDC--------->jdbc
public class mysqltomysql {


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        MySqlSource<String> mySalSource = MySqlSource.<String>builder()
                .hostname("hadoop103")
                .port(3306)
                .databaseList("db01")
                .tableList("db01.tt1")
                .username("root")
                .password("123456")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        DataStreamSource<String> mySQL_source = env.fromSource(mySalSource, WatermarkStrategy.noWatermarks(), "MySQL Source");
        mySQL_source.print();
        SingleOutputStreamOperator<String> after = mySQL_source.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                String after = jsonObject.getString("after");
                System.out.println("-------"+after);
                ToMysql(after);
                return after;
            }

            private void ToMysql(String after) {
                JSONObject jsonObject = JSONObject.parseObject(after);
                Connection connection = null;
                PreparedStatement pstmt = null;
                try {
                    Class.forName("com.mysql.jdbc.Driver");
                    connection = DriverManager.getConnection("jdbc:mysql://hadoop103:3306/db01", "root", "123456");
                    String sql = "insert into t1  values (?,?)";
                    pstmt = connection.prepareStatement(sql);
                    pstmt.setInt(1,jsonObject.getInteger("id"));
                    pstmt.setString(2,jsonObject.getString("name"));
                    pstmt.executeUpdate();
                    pstmt.close();

                } catch (Exception e) {
                    System.out.println("插入失败");
                    e.printStackTrace();
                } finally {
                    if(pstmt != null){
                        try {
                            pstmt.close();
                        } catch (SQLException e) {
                            e.printStackTrace();
                        }
                    }
                    if(connection != null){
                        try {
                            connection.close();
                        } catch (SQLException e) {
                            e.printStackTrace();
                        }
                    }
                }

            }
        });





        env.execute();




    }







}
